package com.johnguo.beam_test.write;

import java.sql.PreparedStatement;
import java.util.Arrays;
import java.util.List;

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.io.jdbc.JdbcIO;
import org.apache.beam.sdk.io.jdbc.JdbcIO.PreparedStatementSetter;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.io.jdbc.JdbcIO.Write;

public class JdbcWriteBeam {
	static final List<String> LINES1 = Arrays.asList("johnguo 30", "johnguo 20", "annasui 40");

	public static void main(String... args) {

		PipelineOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().create();
		Pipeline p = Pipeline.create(options);
		JdbcIO.DataSourceConfiguration config = JdbcIO.DataSourceConfiguration
				.create("com.mysql.jdbc.Driver", "jdbc:mysql://192.168.1.181:3306/johnguo").withUsername("root")
				.withPassword("123456");
		Write<String[]> writer = JdbcIO.<String[]>write().withDataSourceConfiguration(config)
				.withStatement("insert into test_beam values(?,?)")
				.withPreparedStatementSetter(new PreparedStatementSetter<String[]>() {
					private static final long serialVersionUID = 1L;

					@Override
					public void setParameters(String[] element, PreparedStatement ps) throws Exception {
						ps.setObject(1, element[0]);
						ps.setString(2, element[1]);
					}
				});
		p.apply(Create.of(LINES1)).setCoder(StringUtf8Coder.of()).apply(ParDo.of(new DoFn<String, String[]>() {
			private static final long serialVersionUID = 1L;

			@ProcessElement
			public void processElement(ProcessContext c) {
				String[] record = c.element().split(" ");
				c.output(record);
			}
		})).apply(writer);
		p.run().waitUntilFinish();
	}
}
